package edu.utexas.mapreduce;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.URL;
import java.net.URLConnection;
import java.util.jar.JarInputStream;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.codec.binary.Base64;
import org.xml.sax.SAXException;

public class ClientMapReduce {
	
	private CommandLine cmdLine;
	
	public ClientMapReduce(CommandLine cmdline) {
		cmdLine = cmdline;
	}

	public void run() {
		JobRequest newJobRequest = null;
		Job newJob = null;
		//if a jobfile is specified
		if (cmdLine.hasOption(CommandLineParameters.JobFile)){
			//request is specified at least partially through an XML file 
			try {
				File xmlFile = new File (cmdLine.getOptionValue(CommandLineParameters.JobFile.trim()));
				newJobRequest = new JobRequest(xmlFile);
				//Validate job request
				if (!newJobRequest.validate()){
					System.out.println("Job request failed validation");
					return;
				}
				//TODO: Doing only first job for now
				newJob = newJobRequest.getJob(0);			
			} catch (IOException e) {
				e.printStackTrace();
				return;
			} catch (SAXException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				return;
			}
		}
		else {
			//request is specified fully through the command line
			newJob = new Job();
		}
		
		//Assign job request values from command line, falling back on those from the XML, if they exist
		newJob.input = setOption(cmdLine, CommandLineParameters.Input, newJob.input);
		newJob.output = setOption(cmdLine, CommandLineParameters.Output, newJob.output);
		newJob.jobURL = setOption(cmdLine, CommandLineParameters.Url, newJob.jobURL);
		newJob.jar = setOption(cmdLine, CommandLineParameters.Jar, newJob.jar);
		newJob.map = setOption(cmdLine, CommandLineParameters.Map, newJob.map);
		newJob.reduce = setOption(cmdLine, CommandLineParameters.Reduce, newJob.reduce);
		newJob.name = setOption(cmdLine, CommandLineParameters.JobName, newJob.name);
		newJob.requesterName = setOption(cmdLine, CommandLineParameters.RequesterName, newJob.requesterName);
		newJob.requesterEmail = setOption(cmdLine, CommandLineParameters.RequesterEmail, newJob.requesterEmail);
		
		if (!newJob.validate()){
			System.out.println("Job failed validation");
			return;
		}
	
		System.out.println("submit client mapreduce job");
		System.out.println(newJob.toString());

		try {
			// read jar data
			File file = new File(newJob.jar);
			FileInputStream fileIS = new FileInputStream(file);
			byte[] jarData = new byte[fileIS.available()];
			fileIS.read(jarData);
			
			Base64 base64 = new Base64();
			String jarDataBase64 = base64.encodeToString(jarData);
			newJob.jar= "![CDATA[" + jarDataBase64 + "]]";
			//
			// Check input jar before sending to server
			//
			InputStream byteStream = new ByteArrayInputStream(jarData);
			JarClassLoader loader = new JarClassLoader(new JarInputStream(byteStream));
			Class<?> mapClass = loader.loadClass(newJob.map);
			Class<?> reduceClass = loader.loadClass(newJob.reduce);
			
			// check that the classes specified can be used as mapper and reducer
			if (!Mapper.class.isAssignableFrom(mapClass)) {
				System.out.println("Error.  Map class " + newJob.map + " does not implement " + Mapper.class.getName());
			}
			if (!Reducer.class.isAssignableFrom(reduceClass)) {
				System.out.println("Error.  Reduce class " + newJob.reduce + " does not implement " + Reducer.class.getName());
			}
			
			// check that we can instantiate the map/reduce classes 
			mapClass.newInstance();
			reduceClass.newInstance();
			
			File inputFile = new File(newJob.input);
			File outputFile = new File(newJob.output);
			
			if (!inputFile.canRead()) {
				System.out.println("ERROR.  input path is not readable");
				return;
			}
			
			if (!outputFile.canWrite()) {
				System.out.println("ERROR.  output path is not writable");
				return;
			}
			
			
			System.out.println("Send job to mapreduce server");
			URL server = new URL(newJob.jobURL);
			URLConnection serverConnection = server.openConnection();
			serverConnection.setDoOutput(true);
			PrintWriter serverOut = new PrintWriter(serverConnection.getOutputStream());

			serverOut.println(newJob.toXMLString(true));	
			serverOut.flush();
//			serverOut.close();
			
			// Get the server response
			BufferedReader serverIn = new BufferedReader(new InputStreamReader(serverConnection.getInputStream()));
		    String line;
		    while ((line = serverIn.readLine()) != null) {
		        System.out.println(line);
		    }
		    serverOut.close();
		    serverIn.close();
			
		} catch (IOException e1) {
			e1.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
	}
	
	public static String setOption(CommandLine cmdline, String optionName, String defaultValue){
		return cmdline.hasOption(optionName)?cmdline.getOptionValue(optionName):defaultValue;
	}
	
	
}
